package cn.stylefeng.guns.modular.gridsystem.mqtt;

import cn.stylefeng.guns.modular.gridsystem.config.MqttClientConfig;
import cn.stylefeng.guns.modular.gridsystem.entity.BizGateway;
import cn.stylefeng.guns.modular.gridsystem.service.BizGatewayService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import javax.annotation.Resource;
import java.util.List;

/**
 * @author ssy
 * @date 2023/3/7 13:48
 */
@Slf4j
public class MyMqttClient {

    private MyMqttClientCallback mqttClientCallback;

    /**
     * MQTT异步客户端
     */
    private MqttAsyncClient client = null;
    /**
     * 连接配置
     */
    private MqttConnectOptions options = null;

    @Resource
    private MqttClientConfig mqttClientConfig;

    @Resource
    private BizGatewayService gatewayService;

    /**
     * 连接MQTT服务器
     */
    public synchronized void connect() {
        if (options == null) {
            setOptions();
        }
        if (client == null) {
            creatClient();
        }
        int connectTimes = 0;
        while (connectTimes < mqttClientConfig.getMaxConnectTimes() && !client.isConnected()) {
            try {
                IMqttToken token = client.connect(options);
                token.waitForCompletion();
                connectTimes++;
            } catch (Exception e) {
                e.printStackTrace();
                log.info("{} 连接时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

    /**
     * 断开与MQTT服务器的连接
     */
    public synchronized void disconnect() {
        if (client != null && client.isConnected()) {
            try {
                IMqttToken token = client.disconnect();
                token.waitForCompletion();
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("{} 断开连接时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
        client = null;
    }

    /**
     * 刷新MQTT的连接
     */
    public synchronized void refresh() {
        disconnect();
        setOptions();
        creatClient();
        connect();
    }

    /**
     * 共享消息订阅
     */
    public void sharedSubscribe() {
        if (client != null && client.isConnected()) {
            try {
                String[] topics = new String[mqttClientConfig.getTopics().length];
                for (int i = 0; i < mqttClientConfig.getTopics().length; i++) {
                    // topics[i] = "$queue/" + mqttClientConfig.getTopics()[i];
                }
                //EMQ共享订阅模式
                IMqttToken token = client.subscribe(topics, mqttClientConfig.getQos());
                token.waitForCompletion();
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("{} 共享订阅主题时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

    /**
     * 消息订阅
     */
    public void subscribe() {
        if (client != null && client.isConnected()) {
            try {
                //mx
                List<BizGateway> list = gatewayService.list();
                for (BizGateway bizGateway : list) {
                    IMqttToken token3 = client.subscribe("CONTROL/" + bizGateway.getGatewayCode(), 2);
                    token3.waitForCompletion();
                    //mx


                /*    String gatewayIp = bizGateway.getGatewayIp();
                    String deviceTopic = "deviceControl/" + gatewayIp;
                    String similarityTopic = "similarityControl/" + gatewayIp;

                    IMqttToken token = client.subscribe(deviceTopic, 2);
                    IMqttToken token1 = client.subscribe(similarityTopic, 2);
                    token.waitForCompletion();
                    token1.waitForCompletion();*/
                }
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("{} 消息订阅时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

    /**
     * 消息推送
     *
     * @param topic   消息的主题名
     * @param message 消息报文
     */
    public void publish(String topic, String message) {
        if (client != null && client.isConnected()) {
            try {
                log.info("[MQTT]--{}--发送数据：{}", topic, message);
                MqttMessage mqttMessage = new MqttMessage();
                mqttMessage.setPayload(message.getBytes());
                IMqttDeliveryToken token = client.publish(topic, mqttMessage);
                token.waitForCompletion();
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("{} 订阅主题时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

    /**
     * 消息推送
     *
     * @param topic   消息的主题名
     * @param message 消息报文
     */
    public void publish(String topic, byte[] message) {
        if (client != null && client.isConnected()) {
            try {
                IMqttDeliveryToken token = client.publish(topic, message, 2, false);
                token.waitForCompletion();
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("{} 推送消息时发生错误： {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

    /**
     * @return 是否处于连接状态
     */
    public boolean isConnected() {
        return client != null && client.isConnected();
    }

    public String getClientid() {
        return mqttClientConfig.getClientId();
    }

    public int getMaxConnectTimes() {
        return mqttClientConfig.getMaxConnectTimes();
    }

    public void setPushCallback(MyMqttClientCallback myMqttClientCallback) {
        this.mqttClientCallback = myMqttClientCallback;
    }

    /**
     * 设置连接属性
     */
    private void setOptions() {
        if (options != null) {
            options = null;
        }
        options = new MqttConnectOptions();
        //将CleanSession设置为true时，一旦客户端断开连接，就会清除相关Session
        options.setCleanSession(false);
        options.setConnectionTimeout(mqttClientConfig.getTimeOut());
        options.setKeepAliveInterval(mqttClientConfig.getAliveTime());
        //todo
        options.setUserName(mqttClientConfig.getUserName());
      //  options.setPassword(mqttClientConfig.getPassword().toCharArray());
    }

    /**
     * 创建客户端
     */
    private void creatClient() {
        if (client == null) {
            try {
                // host为主机名，mqttClientConfig.getClientId()即连接MQTT的客户端ID，一般以唯一标识符表示，MemoryPersistence设置mqttClientConfig.getClientId()的保存形式，默认为以内存保存
                client = new MqttAsyncClient(mqttClientConfig.getBrokerUrl(), mqttClientConfig.getClientId(), new MemoryPersistence());
                // 设置回调函数
                client.setCallback(new MyMqttClientCallback(MyMqttClient.this));
            } catch (MqttException e) {
                e.printStackTrace();
                log.info("创建连接客户端实例: [{}] 时发生错误: {}", mqttClientConfig.getClientId(), e.toString());
            }
        }
    }

}
